package com.atguigu.bigdata.chapter07.watermark;

import com.atguigu.bigdata.bean.WaterSensor;
import com.atguigu.bigdata.util.AtguiguUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;

/**
 * @Author lzc
 * @Date 2022/9/4 8:28
 */
public class Flink03_WMK_Multi {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        // 今天测试水印的时候: 并行度必须设置 1
        env.setParallelism(2);
        
        // 抽象时间时间
        // 返回值就是这个与元素的事件时间
        env
            .socketTextStream("hadoop162", 9999)
            .map(new MapFunction<String, WaterSensor>() {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
                }
            })
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    // 返回值必须是毫秒. 是一个绝对时间
                    .withTimestampAssigner((element, ts) -> element.getTs())
                    // 如果一个并行度10s每个更新数据(没有更新水印), 水印传递就以其他的为准
                    .withIdleness(Duration.ofSeconds(10))
            )
            .keyBy(WaterSensor::getId)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                
                // 处理窗口中元素的方法: 当窗口关闭的时候执行一次
                @Override
                public void process(String key,  // 属于哪个key的窗口
                                    Context ctx,  // 上下文对象, 可以获取到窗口信息
                                    Iterable<WaterSensor> elements, // 窗口触发计算的时候, 存储着窗口内所有的元素
                                    Collector<String> out) throws Exception {
                    ArrayList<WaterSensor> list = new ArrayList<>();
                    // Iterable很多操作不能做, 比如排序, 一般是把他转成一个List集合再操作
                    for (WaterSensor ws : elements) {
                        list.add(ws);
                    }
                    
                    // 获取窗口的开始时间
                    String stt = AtguiguUtil.toDatTime(ctx.window().getStart());
                    // 获取窗口的结束时间
                    String edt = AtguiguUtil.toDatTime(ctx.window().getEnd());
                    
                    out.collect(key + " " + stt + " " + edt + "  " + list);
                }
            })
            .print();
        
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
/*
生成水印:
    每个并行度根据当前并行度内最大打包时间戳计算得到
    
水印传递:
    广播传递: 从收到的所有水印中找一个最新的传递到后面的所有并行度
        找到的是最小的水印进行传递
 
 */